package com.tmsb.sink;

import com.tmsb.pojo.FairShopOrderMall;
import com.tmsb.utils.ConfUtil;
import com.tmsb.utils.RedisCluster;
import com.tmsb.utils.Util;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.JedisCluster;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;
/**
 * Description: 用tmsb_fair_shop的order_mall来获取每日sku总数这指标的sink
 * Version:1.0.0
 *
 * @author wxw
 * @date 2020/7/27
 */

public class OrderMallSink extends RichSinkFunction<FairShopOrderMall> implements CheckpointedFunction {
    //TODO 总数量
    private ListState<Long> totalSkuQuantityState;
    private Long total_sku;
    //TODO 订单id集合,用于判断是否重复算过
    private ListState<Set> idState;
    private Set idSet;
    private ListState<Long> dateState;
    private Long dateValue;
    private SimpleDateFormat sdf;
    private JedisCluster jedis;
    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        totalSkuQuantityState.clear();
        dateState.clear();
        idState.clear();
        totalSkuQuantityState.add(total_sku);
        dateState.add(dateValue);
        idState.add(idSet);
    }

    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        total_sku = 0L;
        dateValue = 0L;
        idSet = new HashSet();
        jedis = RedisCluster.getInstance(ConfUtil.getRedisCluster()).getRedisClusterList().get(0);
        sdf = new SimpleDateFormat("yyyyMMdd");

        ListStateDescriptor<Set> idSetSkuStateDes =
                new ListStateDescriptor<>("idSetSkuState", TypeInformation.of(new TypeHint<Set>() {}));
        ListStateDescriptor<Long> totalSkuStateDes =
                new ListStateDescriptor<>("totalSkuState", TypeInformation.of(new TypeHint<Long>() {}));
        ListStateDescriptor<Long> om_dateDes =
                new ListStateDescriptor<>("om_date", TypeInformation.of(new TypeHint<Long>() {}));

        idState = context.getOperatorStateStore().getListState(idSetSkuStateDes);
        totalSkuQuantityState = context.getOperatorStateStore().getListState(totalSkuStateDes);
        dateState = context.getOperatorStateStore().getListState(om_dateDes);

        if (context.isRestored()) {
            total_sku = totalSkuQuantityState.get().iterator().next();
            dateValue = dateState.get().iterator().next();
            idSet = idState.get().iterator().next();

        }
    }

    @Override
    public void invoke(FairShopOrderMall value, Context context) throws Exception {
        long orderTime = value.getOrder_time()-28800000;
        //TODO 每天0时清空状态
        if(orderTime>=dateValue){
            idSet = new HashSet();
            total_sku = 0L;
            totalSkuQuantityState.clear();
            idState.clear();
            dateValue = Util.getTomorrowZeroPointTimestamps();
        }else if(orderTime < Util.getZeroPointTimestamps()){
            return;
        }
        //TODO 其余的就是今天的订单，计算每日进货数
        long id = value.getId();
        long sku_quantity ;
      /*  Iterator<Set> iterator = idState.get().iterator();
        while (iterator.hasNext()){
            idSet = iterator.next();
        }*/
        //TODO 已付款，待配送
        if(value.getStatus()==1){
            //TODO 如果订单状态为1的已经来过一次，不要重复计算
            if(idSet.contains(id)){
                return;
            }
            //TODO 新增订单
            sku_quantity = value.getSku_quantity();
            idSet.add(id);
        }else {
            //TODO 订单状态为3，即退款
            if(!idSet.contains(id)){
                return;
            }
            sku_quantity = value.getSku_quantity()*(-1);
            idSet.remove(id);
        }
        total_sku += sku_quantity;

        String date = sdf.format(new Date(orderTime));
        //TODO 向redis发送数据
        jedis.hset(ConfUtil.getOrderMallRedisKey(),date,total_sku+"");

    }
}
